package org.rambadger.command.axon;

import com.mongodb.MongoClient;
import com.yammer.dropwizard.ConfiguredBundle;
import com.yammer.dropwizard.config.Bootstrap;
import com.yammer.dropwizard.config.Environment;
import org.axonframework.commandhandling.CommandBus;
import org.axonframework.commandhandling.SimpleCommandBus;
import org.axonframework.commandhandling.annotation.AggregateAnnotationCommandHandler;
import org.axonframework.commandhandling.gateway.CommandGateway;
import org.axonframework.commandhandling.gateway.DefaultCommandGateway;
import org.axonframework.commandhandling.interceptors.BeanValidationInterceptor;
import org.axonframework.eventhandling.*;
import org.axonframework.eventhandling.annotation.AnnotationEventListenerAdapter;
import org.axonframework.eventhandling.replay.BackloggingIncomingMessageHandler;
import org.axonframework.eventhandling.replay.ReplayAware;
import org.axonframework.eventhandling.replay.ReplayingCluster;
import org.axonframework.eventsourcing.EventSourcedAggregateRoot;
import org.axonframework.eventsourcing.EventSourcingRepository;
import org.axonframework.eventstore.mongo.DocumentPerCommitStorageStrategy;
import org.axonframework.eventstore.mongo.MongoEventStore;
import org.axonframework.saga.GenericSagaFactory;
import org.axonframework.saga.SimpleResourceInjector;
import org.axonframework.saga.annotation.AnnotatedSagaManager;
import org.axonframework.saga.repository.mongo.MongoSagaRepository;
import org.axonframework.serializer.bson.DBObjectXStreamSerializer;
import org.axonframework.unitofwork.NoTransactionManager;
import org.rambadger.command.CommandConfiguration;
import org.rambadger.command.auth.AuthListener;
import org.rambadger.command.auth.AuthUser;
import org.rambadger.command.listeners.*;
import org.rambadger.command.model.PlatformAggregate;
import org.rambadger.command.model.SubmissionSaga;
import org.rambadger.command.model.UserAggregate;
import org.rambadger.command.providers.SingletonInjectorFactory;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static com.google.common.collect.Lists.newArrayList;

import org.rambadger.command.representation.*;

/**
 * DropWizard bundle that sets up the Axon stuff
 */
public class AxonBundle implements ConfiguredBundle<CommandConfiguration> {
    private final MongoBundle mongo;
    private MongoEventStore eventStore;

    private ReplayingCluster replayingCluster;
    private ClusteringEventBus eventBus;

    private CommandBus commandBus;
    private CommandGateway gateway;
    private Map<Class<?>, EventSourcingRepository<?>> repositories = new HashMap<>();

    //Sagas
    private MongoSagaRepository sagaRepository;
    private GenericSagaFactory sagaFactory;
    private SimpleResourceInjector resourceInjector;
    private AnnotatedSagaManager sagaManager;

    /**
     * Instantiates a new AxonBundle instance.
     *
     * @param mongo the MongoBundle instance
     */
    public AxonBundle(MongoBundle mongo) {
        this.mongo = mongo;
    }


    @Override
    public void initialize(Bootstrap<?> bootstrap) {
        //this does NOT add MongoBundle
    }

    @Override
    public void run(CommandConfiguration configuration, Environment environment) throws Exception {
        MongoConfiguration conf = configuration.getDatabaseConfiguration();
        MongoClient client = mongo.getMongoClient();

        this.eventStore = new MongoEventStore(conf.getMongoEventTemplate(client), new DBObjectXStreamSerializer(), new DocumentPerCommitStorageStrategy());

        //Clusters have to be set up before the event bus
        this.replayingCluster = new ReplayingCluster(new SimpleCluster("replay"), eventStore,
                                                     //Mongo doesn't do transactions
                                                     new NoTransactionManager(), 0,
                                                     new BackloggingIncomingMessageHandler());

        /*
        Since only replay compatible event handlers should be put on the replay cluster (bad things can happen). We need to have a separate clusters for compatible and non compatible event handlers.
         */
        ClusterSelector replaySelector = new SuperclassClusterSelector(ReplayAware.class, replayingCluster);
        ClusterSelector defaultSelector = new DefaultClusterSelector();
        //NOTE: The the order of the selectors passed to the CompositeClusterSelector is important
        List<ClusterSelector> clusterSelectors = newArrayList(replaySelector, defaultSelector);
        this.eventBus = new ClusteringEventBus(new CompositeClusterSelector(clusterSelectors));

        //Command Bus/Gateway
        this.commandBus = new SimpleCommandBus();
        this.gateway = new DefaultCommandGateway(commandBus, new BeanValidationInterceptor());
        environment.addProvider(SingletonInjectorFactory.getProvider(CommandGateway.class, gateway));

        //Repositories
        setupRepository(UserAggregate.class);
        setupRepository(PlatformAggregate.class);

        //Listeners
        AnnotationEventListenerAdapter.subscribe(new AuthListener(mongo.getWrappedCollection(AuthUser.class)), eventBus);
        AnnotationEventListenerAdapter.subscribe(new UserListener(mongo.getWrappedCollection(User.class)), eventBus);
        AnnotationEventListenerAdapter.subscribe(new PlatformListener(mongo.getWrappedCollection(Platform.class)), eventBus);
        AnnotationEventListenerAdapter.subscribe(new TrackListener(mongo.getWrappedCollection(Track.class, Track.TrackId.class)), eventBus);
        AnnotationEventListenerAdapter.subscribe(new SubmissionListener(mongo.getWrappedCollection(Submission.class, Submission.SubmissionId.class)), eventBus);
        AnnotationEventListenerAdapter.subscribe(new AssignmentListener(mongo.getWrappedCollection(Assignment.class, Assignment.AssignmentId.class)), eventBus);



        //Sagas
        resourceInjector = new SimpleResourceInjector(gateway);

        sagaRepository = new MongoSagaRepository(conf.getMongoSagaTemplate(client));
        sagaRepository.setResourceInjector(resourceInjector);

        sagaFactory = new GenericSagaFactory();
        sagaFactory.setResourceInjector(resourceInjector);
        //Additional Saga types can be added to the varargs
        sagaManager = new AnnotatedSagaManager(sagaRepository, sagaFactory, eventBus, SubmissionSaga.class);
        sagaManager.subscribe();
    }

    /**
     * Gets the command gateway.
     *
     * @return the command gateway
     */
    public CommandGateway getCommandGateway() {
        return gateway;
    }

    /**
     * Gets the replaying cluster.
     *
     * @return the replaying cluster
     */
    public ReplayingCluster getReplayingCluster() {
        return replayingCluster;
    }

    /**
     * Returns a configured {@link EventSourcingRepository } for the given aggregate type.
     * <p>
     *     It either retrieves a new Repository from the map of already created ones, or constructs a new one.
     * </p>
     * @param type The class to construct a Repository for
     * @param <T> The type of the repository
     * @return a configured repository
     */
    @SuppressWarnings("unchecked")
    private <T extends EventSourcedAggregateRoot> EventSourcingRepository<T> setupRepository(Class<T> type) {
        if (this.repositories.containsKey(type)) {
            return (EventSourcingRepository<T>) this.repositories.get(type);
        } else {
            EventSourcingRepository<T> repository = new EventSourcingRepository<>(type, this.eventStore);
            repository.setEventBus(this.eventBus);
            AggregateAnnotationCommandHandler.subscribe(type, repository, this.commandBus);
            this.repositories.put(type, repository);
            return repository;
        }
    }
}
